#include "pio.h"
#include "sockio.h"

#include <stdlib.h>
#include <string.h>
#include <sys/socket.h>
#include <pthread.h>
#include <assert.h>
#include <stdio.h>
#include <unistd.h>
#include <fcntl.h>
#include <errno.h>

typedef enum
{
  WORK_READ,
  WORK_WRITE
} worktype;

/* estrutura mestre de trabalhadores */
struct pio_t {
  int                   id; /* sockfd */
  struct worker_t       *workers; /* worker pool */
  struct work_t         *todo; /* work-to-do list */
  
  pthread_mutex_t       workwait; /* lock do workstart */
  pthread_cond_t        workstart; /* sinaliza que ha trabalho a fazer */
  pthread_t             overseer; /* thread sinalizadora e que recebe */
  unsigned int          poolsize; /* numero maximo de threads */
};

/* dados privados de cada thread trabalhadora */
typedef struct worker_t {
  int                    id; /* thread id */
  pthread_t              thread;
  
  struct pio_t           *master;
  
  pthread_mutex_t        *workwait; /* lock do workstart */
  pthread_cond_t         *workstart; /* sinaliza que ha trabalho a fazer */  
  
} worker;

/* um trabalho de io a ser feito */
typedef struct work_t {
  char                   *bufp; /* buffer de trabalho da thread */
  size_t                 bufsize; /* tamanho do buffer da thread */
  FILE                   *file; /* fonte dos dados */
  size_t                 total; /* total de bytes efetivamente usados */
  
  struct work_t          *next; /* proximo trabalho a fazer */
  
  worktype               type; /* tipo: read ou write */
  
  piocallback            done; /* done callback */
  void                   *priv; /* dados privados */
  
} work;
  
/* thread de trabalho */
void *dowork(void *data_);

/* sinaliza work to do */
void *overseer(void *io_)
{
  pio *io = io_;
  work * workp = 0;
  
  assert(io != NULL);
  
  
  printf("Overseer thread is here.\n");
  
  while (1)
  {
    /* notifica work-complete */
    while (sockread(io->id, (void*)&workp, sizeof(&workp)) > 0)
    {
//       printf("workdone: %x, %x->done = %x\n", workp, workp, workp->done);
      workp->done(workp->priv, workp->total);
      free(workp);
    }
    
    
    
    if (io->todo == NULL)
    {
      usleep(500); /* aquela-aliviada */
      continue;
    }
    
    pthread_mutex_lock(&(io->workwait)); /* locka */
    pthread_cond_signal(&(io->workstart)); /* sinaliza trabalho */
    pthread_mutex_unlock(&(io->workwait)); /* deslocka */
  }
}

/* adiciona um trabalho na lista */
void addwork(pio *io, work *wk)
{
  assert(io != 0);
  assert(wk != 0);
  
  int i = 0;
  
  work *workp;
  
  pthread_mutex_lock(&(io->workwait));
  
  if (io->todo == NULL)
  {
    io->todo = wk;
  }
  else
  {
    for (i = 0, workp = io->todo; workp->next != NULL; workp = workp->next, i++);
    workp->next = wk;
  }
  
//   fprintf(stderr, "Adding work %x at position %d\n", wk, i);
  pthread_mutex_unlock(&(io->workwait));
}

/* cria um pio-master */
pio *piocreate (size_t poolsize)
{
  unsigned int i = 0;
  worker *w = 0;
  
  if (poolsize == 0)
    return 0;
  
  /* id = socket; */
  
  pio *io = calloc(sizeof(pio), 1);
  io->todo = NULL;
  
  pthread_attr_t detach;
  pthread_attr_init(&detach);
  pthread_attr_setdetachstate(&detach, PTHREAD_CREATE_DETACHED);
  
  io->id = socklisten(9998, 1, SOCKET_UDP);
  if (io->id < 0)
  {
    free(io);
    perror("Erro ao criar parallel io");
    return 0;
  }
  
  fcntl(io->id, F_SETFL, O_NONBLOCK);
  
  io->poolsize = poolsize;
  io->workers = calloc(sizeof(worker), poolsize);
  
  pthread_mutex_init(&(io->workwait), NULL);
  pthread_cond_init (&(io->workstart), NULL);
  
  for (i = 0, w = io->workers; i < poolsize; i++)
  {
    w = &(io->workers[i]);
    w->id = i;
    w->workwait = &(io->workwait);
    w->workstart = &(io->workstart);
    w->master = io;
    pthread_create(&(w->thread), &detach, dowork, w);
  }
  
  /* cria a thread de sinalizacao */
  pthread_create(&(io->overseer), &detach, overseer, io);
  
  pthread_attr_destroy(&detach);
  
  return io;
}

int pioupdate (pio *io)
{
  return 0; // FIXME DEPRECATED TODO
}


/* destroi um pio */
void piodestroy (pio *io)
{
  work *wk;
  size_t i = 0;
  assert(io != 0);

  printf("pio quit\n");
  
  /* destroi a overseer */
  pthread_cancel(io->overseer);
  pthread_detach(io->overseer);
  
  /* destroi as threads de trabalho */
  for (i = 0; i < io->poolsize; i++)
  {
    pthread_cancel(io->workers[i].thread);
    pthread_detach(io->workers[i].thread);
  }
  
  free(io->workers);
  
  sockclose(io->id);
  
  pthread_cond_destroy(&(io->workstart));
  pthread_mutex_destroy(&(io->workwait));
  
  /* apaga a lista de trabalhos restantes */
  while (io->todo != NULL)
  {
    wk = io->todo;
    io->todo = wk->next;
    
    free(wk);
  }
  
  free(io);
}


void *dowork(void *data_)
{
  worker *self = (worker *) data_;
  pio *master = self->master;
  unsigned int id = self->id;
  work *todo = 0; 
  size_t status = 0;
  
  printf("workthread %d boot.\n", id);
  
  /* work loop */
  while (1)
  {
    pthread_mutex_lock(self->workwait);
//     printf("workthread %d will wait for work.\n",id);
    pthread_cond_wait(self->workstart, self->workwait);
    
    /* spurious wake-up */
    if (master->todo == NULL)
    {
//       printf("workthread %d called but nothing to do. Stupid SMP\n", id);
      pthread_mutex_unlock(self->workwait);
      continue;
    }
    
    todo = master->todo;
    master->todo = todo->next;
//     printf("workthead %d acquiring job. Next is: %x\n", id, todo->next);
    pthread_mutex_unlock(self->workwait); 
    
//     printf("workthread %d will do\n", id);
    while (todo->total < todo->bufsize)
    {
      
//       printf("filepos: %ld bytes, %d total\n", ftell(todo->file), todo->total);
      
      if (todo->type == WORK_WRITE)
        status = fwrite(todo->bufp + todo->total, sizeof(char), todo->bufsize - todo->total, todo->file);
      else
        status = fread (todo->bufp + todo->total, sizeof(char), todo->bufsize - todo->total, todo->file);
      
      if (status < 0)
      {
        /* error */
        fprintf(stderr, "workthread %d thinks the file is buggy: %s\n", id, strerror(errno)); 
        break;
      }
      
      todo->total += status;
      
      /* maybe eof */
      if (status == 0)
      {
//         fprintf(stderr, "workthread %d total = %d bufsize = %d eof = %d\n", id, todo->total, todo->bufsize - todo->total, feof(todo->file));
        break;
      }
    }
//     printf("workthread %d finished\n", id);
    int fd = sockopen("localhost", 9998, SOCKET_UDP);
    sockwrite(fd, (char*)&todo, sizeof(&todo));
    sockclose(fd);
    
    usleep(500);
  }
}

void pioread(pio *io, FILE *from, void *to, size_t size, void *data, piocallback done)
{
  work  *w;
  
  assert(io != 0);
  assert(to != 0);
  assert(from != 0);
  if (size == 0) return;
  
  w = calloc(sizeof(work), 1);
  w->bufp = to;
  w->bufsize = size;
  w->file = from;
  w->done = done;
  w->priv = data;
  w->total = 0;
  w->type = WORK_READ;
  
  addwork(io, w);
}

void piowrite(pio *io, FILE *to, void *from, size_t size, void *data, piocallback done)
{
  work *w;
  
  assert(io != 0);
  assert(to != 0);
  assert(from != 0);
  if (size == 0) return;
  
  w = calloc(sizeof(work), 1);
  w->bufp = from;
  w->bufsize = size;
  w->file = to;
  w->done = done;
  w->priv = data;
  w->total = 0;
  w->type = WORK_WRITE;
  
  addwork(io, w);
}

